package com.atguigu.ad.hive.udf;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.*;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.io.IOUtils;
import org.lionsoul.ip2region.xdb.Searcher;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;

/**
 *      hive自定义函数：
 * 解决      通过ip地址查询出ip所处的地址，返回的是结构体
 * 前提    https://github.com/lionsoul2014/ip2region.git    去下载 ip2region.xdb  保存到HDFS
 */
public class ParseIP extends GenericUDF {

    private Searcher searcher;
    @Override
    public ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException {

        //        TODO 判断参数是否正确
        if(objectInspectors.length != 2){
            throw new UDFArgumentException("parse_ip函数需要接受两个参数");
        }
//       TODO 判断第一个参数是否正确
        ObjectInspector hdfsPathOI = objectInspectors[0];
        if(hdfsPathOI.getCategory() != ObjectInspector.Category.PRIMITIVE){
            throw new UDFArgumentException("parse_ip函数的第1个参数应为基本数据类型");
        }

        PrimitiveObjectInspector hdfsPathOI1 = (PrimitiveObjectInspector) hdfsPathOI;
        if(hdfsPathOI1.getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.STRING){
            throw new UDFArgumentException("parse_ip函数的第1个参数应为STRING类型");
        }


        //        TODO 获取hdfs中的包，加载到内存
        if(hdfsPathOI instanceof ConstantObjectInspector) {
//            获取第一个参数（HDFS路径）的值（常量）
          String filePath =   ((ConstantObjectInspector) hdfsPathOI).getWritableConstantValue().toString();
//            创建path对象
          Path path = new Path(filePath);
            Configuration conf = new Configuration();
//            获取HDFS文件系统客服端
            try {
                FileSystem fs =  FileSystem.get(conf);
//            打开文件获取输入流
                FSDataInputStream inputStream = fs.open(path);
                ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
                IOUtils.copyBytes(inputStream,outputStream,1024);
                byte[] bytes = outputStream.toByteArray();
                searcher = Searcher.newWithBuffer(bytes);
            } catch (IOException e) {
                e.printStackTrace();
            }
//            TODO 检查第二个参数
            ObjectInspector objectInspector = objectInspectors[1];
            if(objectInspector.getCategory() != ObjectInspector.Category.PRIMITIVE){
                throw new UDFArgumentException("parse_ip函数的第2个参数应为基本数据类型");
            }

            PrimitiveObjectInspector primitiveIPOI = (PrimitiveObjectInspector) objectInspector;
            if(primitiveIPOI.getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.STRING){
                throw new UDFArgumentException("parse_ip函数的第2个参数应为STRING类型");
            }

        }
//        TODO 返回的是结构体  结构体的定义
        ArrayList<String> structFieldNames = new ArrayList<>();
        ArrayList<ObjectInspector> structFileldObctInspectors = new ArrayList<>();

        structFieldNames.add("country");
        structFieldNames.add("area");
        structFieldNames.add("province");
        structFieldNames.add("city");
        structFieldNames.add("isp");

        structFileldObctInspectors.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        structFileldObctInspectors.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        structFileldObctInspectors.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        structFileldObctInspectors.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        structFileldObctInspectors.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);



        return ObjectInspectorFactory.getStandardStructObjectInspector(structFieldNames,structFileldObctInspectors);
    }

    @Override
    public Object evaluate(DeferredObject[] deferredObjects) throws HiveException {
        String ipAddress = deferredObjects[1].get().toString();
        ArrayList<Object> result = new ArrayList<>();

        try {
//            去包中查询地址，返回|的分割
            String region = searcher.search(ipAddress);
            String[] split = region.split("\\|");

            result.add(split[0]);
            result.add(split[1]);
            result.add(split[2]);
            result.add(split[3]);
            result.add(split[4]);
        } catch (Exception e) {
            e.printStackTrace();
        }

        return result;
    }

    @Override
    public String getDisplayString(String[] strings) {
//        函数名的定义
        return getStandardDisplayString("parse_ip",strings);
    }
}
